/**
 * 
 */
package qy.jalgotrade.optimizer.local;

import java.io.IOException;
import java.net.MalformedURLException;
import java.net.ServerSocket;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.commons.lang3.reflect.ConstructorUtils;
import org.apache.commons.math3.stat.descriptive.summary.Sum;
import org.apache.xmlrpc.XmlRpcException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import ch.qos.logback.classic.Level;
import qy.jalgotrade.barfeed.BaseBarFeed;
import qy.jalgotrade.optimizer.base.ParameterSource;
import qy.jalgotrade.optimizer.base.Parameters;
import qy.jalgotrade.optimizer.base.ResultSincFactory;
import qy.jalgotrade.optimizer.base.AbstractResultSinc;
import qy.jalgotrade.optimizer.base.OptimizingGoal;
import qy.jalgotrade.optimizer.base.Results;
import qy.jalgotrade.optimizer.worker.AbstractWorker;
import qy.jalgotrade.optimizer.xmlrpc.XmlRpcServerRunnable;
import qy.jalgotrade.stratanalyzer.DrawDown;
import qy.jalgotrade.stratanalyzer.SharpeRatio;
import qy.jalgotrade.stratanalyzer.Trades;
import qy.jalgotrade.stratanalyzer.returns.Returns;
import qy.jalgotrade.strategy.BacktestingStrategy;
import qy.jalgotrade.utils.CommonUtils;
import qy.jalgotrade.utils.PySpecs.Iter;

/**
 * @author c-geo
 *
 */
public final class LocalOptimizer {

	public static final String LOCAL_OPTIMIZER_SERVER = "LocalOptimizer-Server";

	/**
	 * 
	 * @author c-geo
	 *
	 */
	private static class LocalWorkerRunnable extends AbstractWorker {

		/**
		 * 
		 */
		private Class<? extends BacktestingStrategy> __strategyClass;

		/**
		 * 
		 * @param strategyClass
		 * @param address
		 * @param port
		 * @throws UnknownHostException
		 * @throws MalformedURLException
		 */
		public LocalWorkerRunnable(Class<? extends BacktestingStrategy> strategyClass, String address, int port,
		        OptimizingGoal goal) throws UnknownHostException, MalformedURLException {

			this(strategyClass, address, port, goal, "");
		}

		/**
		 * 
		 * @param strategyClass
		 * @param address
		 * @param port
		 * @param goal
		 * @param workerName
		 * @throws UnknownHostException
		 * @throws MalformedURLException
		 */
		public LocalWorkerRunnable(Class<? extends BacktestingStrategy> strategyClass, String address, int port,
		        OptimizingGoal goal, String workerName) throws UnknownHostException, MalformedURLException {

			super(address, port, goal, workerName);
			__strategyClass = strategyClass;
		}

		/**
		 * 
		 */
		@Override
		public Results runStrategy(BaseBarFeed feed, Parameters parameters) {

			Class<?>[] argTypes = new Class[parameters.getKwargs().size() + 1];
			Object[] args = new Object[parameters.getKwargs().size() + 1];
			argTypes[0] = BaseBarFeed.class;
			args[0] = feed;

			int i = 1;
			for (Map.Entry<String, Object> kvp : parameters.getKwargs().entrySet()) {
				argTypes[i] = kvp.getValue().getClass();
				args[i] = kvp.getValue();
				i++;
			}

			BacktestingStrategy strat;
			OptimizingGoal goal = getGoal();
			SharpeRatio sharpe = null;
			Returns returns = null;
			Trades trades = null;
			DrawDown drawdown = null;
			try {
				strat = ConstructorUtils.invokeConstructor(__strategyClass, args);
				if (goal != null && goal.getGoalKpi() != Results.KPI.RESULT) {
					sharpe = new SharpeRatio();
					returns = new Returns();
					trades = new Trades();
					drawdown = new DrawDown();
					strat.attachAnalyzerEx(sharpe, "sharpe");
					strat.attachAnalyzerEx(returns, "returns");
					strat.attachAnalyzerEx(drawdown, "drawdown");
					strat.attachAnalyzerEx(trades, "trades");
				}
				strat.run();

			} catch (Exception e) {

				getLogger().error("Exception: {} (stack trace: {})", e, ExceptionUtils.getStackTrace(e));
				return null;
			}

			if (goal == null || goal.getGoalKpi() == Results.KPI.RESULT) {

				return new Results(parameters, strat.getResult());
			} else {

				Sum sum = new Sum();
				Double[] values = new Double[trades.getCommissionsForAllTrades().size()];
				double allCommissions = sum
				        .evaluate(ArrayUtils.toPrimitive(trades.getCommissionsForAllTrades().toArray(values)));

				return new Results(parameters, strat.getResult(), returns.getCumulativeReturns().__getitem__(-1),
				        sharpe.getSharpeRatio(0.035), drawdown.getMaxDrawDown(), drawdown.getLongestDrawDownDuration(),
				        feed.getTotalTradeDays(), trades.getCount(), trades.getProfitableCount(), trades.getUnprofitableCount(),
				        trades.getEvenCount(),
				        trades.getCount() != 0 ? ((double) trades.getProfitableCount()) / trades.getCount()
				                : Double.NaN,
				        allCommissions);
			}
		}
	}

	private static Logger logger = LoggerFactory.getLogger(LocalOptimizer.class);

	/**
	 * Executes many instances of a strategy in parallel and finds the parameters that yield the best
	 * results.
	 * 
	 * @param strategyClass      The strategy class.
	 * @param barFeed            The bar feed to use to backtest the strategy.
	 * @param strategyParameters The set of parameters to use for backtesting. An iterable object where
	 *                           **each element is a tuple that holds parameter values**.
	 *                           strategyParameters: 回测使用的 参数集. 需为 iterable, 每次迭代返回一个 参数组
	 *                           tuple(base.Parameters);
	 * @return
	 * @throws UnknownHostException
	 * @throws XmlRpcException
	 */
	public static Results run(Class<? extends BacktestingStrategy> strategyClass, BaseBarFeed barFeed,
	        Iter<Parameters> strategyParameters) throws UnknownHostException, XmlRpcException {

		return run(strategyClass, barFeed, strategyParameters, new OptimizingGoal(), 0, 200, Level.ERROR);
	}

	/**
	 * Executes many instances of a strategy in parallel and finds the parameters that yield the best
	 * results.
	 * 
	 * @param strategyClass      The strategy class.
	 * @param barFeed            The bar feed to use to backtest the strategy.
	 * @param strategyParameters The set of parameters to use for backtesting. An iterable object where
	 *                           **each element is a tuple that holds parameter values**.
	 *                           strategyParameters: 回测使用的 参数集. 需为 iterable, 每次迭代返回一个 参数组
	 *                           tuple(base.Parameters);
	 * @param goal
	 * @return
	 * @throws UnknownHostException
	 * @throws XmlRpcException
	 */
	public static Results run(Class<? extends BacktestingStrategy> strategyClass, BaseBarFeed barFeed,
	        Iter<Parameters> strategyParameters, OptimizingGoal goal) throws UnknownHostException, XmlRpcException {

		return run(strategyClass, barFeed, strategyParameters, goal, 0, 200, Level.ERROR);
	}

	/**
	 * Executes many instances of a strategy in parallel and finds the parameters that yield the best
	 * results.
	 * 
	 * @param strategyClass      The strategy class.
	 * @param barFeed            The bar feed to use to backtest the strategy.
	 * @param strategyParameters The set of parameters to use for backtesting. An iterable object where
	 *                           **each element is a tuple that holds parameter values**.
	 *                           strategyParameters: 回测使用的 参数集. 需为 iterable, 每次迭代返回一个 参数组
	 *                           tuple(base.Parameters);
	 * @param goal
	 * @param workerCount        The number of strategies to run in parallel. If None then as many
	 *                           workers as CPUs are used.
	 * @return
	 * @throws UnknownHostException
	 * @throws XmlRpcException
	 */
	public static Results run(Class<? extends BacktestingStrategy> strategyClass, BaseBarFeed barFeed,
	        Iter<Parameters> strategyParameters, OptimizingGoal goal, int workerCount)
	        throws UnknownHostException, XmlRpcException {

		return run(strategyClass, barFeed, strategyParameters, goal, workerCount, 200, Level.ERROR);
	}

	/**
	 * Executes many instances of a strategy in parallel and finds the parameters that yield the best
	 * results.
	 * 
	 * @param strategyClass      The strategy class.
	 * @param barFeed            The bar feed to use to backtest the strategy.
	 * @param strategyParameters The set of parameters to use for backtesting. An iterable object where
	 *                           **each element is a tuple that holds parameter values**.
	 *                           strategyParameters: 回测使用的 参数集. 需为 iterable, 每次迭代返回一个 参数组
	 *                           tuple(base.Parameters);
	 * @param goal
	 * @param workerCount        The number of strategies to run in parallel. If None then as many
	 *                           workers as CPUs are used.
	 * @param batchSize          The number of strategy executions that are delivered to each worker.
	 *                           batchSize: 分配给每个 worker 参数组 的大小;
	 * @return
	 * @throws UnknownHostException
	 * @throws XmlRpcException
	 */
	public static Results run(Class<? extends BacktestingStrategy> strategyClass, BaseBarFeed barFeed,
	        Iter<Parameters> strategyParameters, OptimizingGoal goal, int workerCount, int batchSize)
	        throws UnknownHostException, XmlRpcException {

		return run(strategyClass, barFeed, strategyParameters, goal, workerCount, batchSize, Level.ERROR);
	}

	/**
	 * Executes many instances of a strategy in parallel and finds the parameters that yield the best
	 * results.
	 * 
	 * @param strategyClass      The strategy class.
	 * @param barFeed            The bar feed to use to backtest the strategy.
	 * @param strategyParameters The set of parameters to use for backtesting. An iterable object where
	 *                           **each element is a tuple that holds parameter values**.
	 *                           strategyParameters: 回测使用的 参数集. 需为 iterable, 每次迭代返回一个 参数组
	 *                           tuple(base.Parameters);
	 * @param goal
	 * @param workerCount        The number of strategies to run in parallel. If None then as many
	 *                           workers as CPUs are used.
	 * @param batchSize          The number of strategy executions that are delivered to each worker.
	 *                           batchSize: 分配给每个 worker 参数组 的大小;
	 * @param logLevel           The log level. Defaults to **logging.ERROR**.
	 * @return
	 * @throws UnknownHostException
	 * @throws XmlRpcException
	 */
	public static Results run(Class<? extends BacktestingStrategy> strategyClass, BaseBarFeed barFeed,
	        Iter<Parameters> strategyParameters, OptimizingGoal goal, int workerCount, int batchSize, Level logLevel)
	        throws UnknownHostException, XmlRpcException {

		if (workerCount <= 0) {
			workerCount = Integer.parseInt(System.getenv(CommonUtils.NUMBER_OF_PROCESSORS));
		}
		assert workerCount > 0 : "No workers";

		Results ret = null;
		List<Thread> workers = new ArrayList<>();
		int port = findPort();
		// Build and start the server thread before the worker processes.
		// We'll manually stop the server once workers have finished.
		ParameterSource paramSource = new ParameterSource(strategyParameters);
		// Create and start the server.
		logger.info("Starting server on port {}", port);

		AbstractResultSinc resultSinc = ResultSincFactory.create(goal);
		XmlRpcServerRunnable server = new XmlRpcServerRunnable(paramSource, resultSinc, barFeed, "localhost", port,
		        goal, false, batchSize);
		Thread serverThread = new Thread(server, LOCAL_OPTIMIZER_SERVER);
		serverThread.start();

		logger.info("Waiting for the server to be ready");

		server.waitServing();

		try {
			logger.info("Starting {} workers", workerCount);
			// Build the worker processes.
			for (int i = 0; i < workerCount; i++) {
				// Create a worker and run it.
				LocalWorkerRunnable wr = new LocalWorkerRunnable(strategyClass, "localhost", port, goal);
				((ch.qos.logback.classic.Logger) LocalWorkerRunnable.getLogger()).setLevel(logLevel);
				Thread workerThread = new Thread(wr);
				wr.changeWorkerName("worker-" + workerThread.getName());
				workers.add(workerThread);
			}
			// Start workers
			for (Thread e : workers) {
				e.start();
			}
			// Wait for all jobs to complete.
			while (server.jobsPending()) {
				Thread.sleep(1000);
			}
		} catch (Exception e) {
			logger.error("Exception: {} (stack trace: {})", e, ExceptionUtils.getStackTrace(e));
		} finally {
			// Stop workers
			for (Thread e : workers) {
				try {
					logger.info("Stopping worker thread {}", e.getName());
					e.join(3000); // TODO: This is necessary to avoid zombie processes.
					if (e.isAlive()) {
						logger.error("worker thread: {} is still alive!", e.getName());
					}
				} catch (InterruptedException e1) {
					logger.error("InterruptedException: {} (stack trace: {})", e1, ExceptionUtils.getStackTrace(e1));
				}
			}

			// Stop and wait the server to finish.
			logger.info("Stopping server");
			server.stop();
			try {
				serverThread.join();
			} catch (InterruptedException e1) {
				logger.error("InterruptedException: {} (stack trace: {})", e1, ExceptionUtils.getStackTrace(e1));
			}

			ret = resultSinc.getBest();
		}

		return ret;
	}

	/**
	 * 随机获取一个空闲的 port ( [1025, 65536) ):
	 * 
	 * @return
	 */
	private static int findPort() {

		int ret;
		Random rnd = new Random(System.currentTimeMillis());
		while (true) {
			try {
				ret = 1025 + rnd.nextInt(65536 - 1025);
				ServerSocket s = new ServerSocket(ret);
				s.close();
				return ret;
			} catch (IOException ioe) {
				// retry
			}
		}
	}

	/**
	 * 
	 */
	private LocalOptimizer() {

	}
}
